use async_channel::unbounded;
use cfgdb::{do_cmd, init_dir, Client, Lock, SvCache};
use lazy_static::lazy_static;
use serde_json::Value;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tracing::{instrument, warn};
use tracing_subscriber::prelude::*;
use warp::{hyper::body::Bytes, Filter};

#[instrument(skip_all, name = "main")]
#[tokio::main]
async fn main() {
    base::trace_warn!();

    init_dir();

    lazy_static! {
        static ref LOCK: Lock = Arc::new(Mutex::new(()));
        static ref CACHE: SvCache = SvCache::builder()
            .max_capacity(90_0000)
            .time_to_idle(Duration::from_secs(24 * 60 * 60 * 15)) // 15 days
            .build();
        static ref CLIENT: Client = Client::new().expect("database init error");
        static ref CHANNEL: (base::CSend, base::CRece) = unbounded::<Value>();
    }

    tokio::spawn(async move {
        let _ = cfgdb::queue_chunk(&CHANNEL.1, &CLIENT).await;
    });

    let md = "cfgdb";
    let body = base::body!();

    let post_three = warp::path(md)
        .and(warp::post())
        .and(warp::path::param())
        .and(warp::path::param())
        .and(body)
        .and_then(move |model: String, cmd: String, data: Bytes| async {
            let lock = Arc::clone(&LOCK);
            do_cmd(
                data,
                model,
                cmd,
                None,
                &CLIENT,
                CACHE.clone(),
                lock,
                CHANNEL.0.clone(),
            )
            .await
        });

    let post_four = warp::path(md)
        .and(warp::post())
        .and(warp::path::param())
        .and(warp::path::param())
        .and(warp::path::param())
        .and(body)
        .and_then(
            move |model: String, cmd: String, name: String, data: Bytes| async {
                let lock = Arc::clone(&LOCK);
                do_cmd(
                    data,
                    model,
                    cmd,
                    Some(name),
                    &CLIENT,
                    CACHE.clone(),
                    lock,
                    CHANNEL.0.clone(),
                )
                .await
            },
        );

    let post = post_four.or(post_three).with(base::cors!());

    let get = warp::get().map(move || format!("\nserve: {}\n\n", md));
    let opt = base::opt!();

    let route = get.or(opt).or(post);

    warn!("listening on 0.0.0.0:9981");

    tokio::spawn(async move {
        warp::serve(route).run(([0, 0, 0, 0], 9981)).await;
    });

    cache_signal(&CLIENT).await;
}

#[cfg(target_os = "linux")]
async fn cache_signal(client: &Client) {
    use futures::stream::StreamExt;
    use signal_hook::consts::signal::*;
    use signal_hook_tokio::Signals;

    let mut signals = Signals::new(&[SIGTERM, SIGINT, SIGQUIT]).expect("register single error");

    while let Some(signal) = signals.next().await {
        match signal {
            SIGTERM | SIGINT | SIGQUIT => {
                println!("cache signal: {}", signal);

                let _ = client.flush();
                break;
            }
            _ => {}
        }
    }
}

#[cfg(not(target_os = "linux"))]
async fn cache_signal(client: &Client) {
    loop {
        use tokio::signal::ctrl_c;

        ctrl_c().await.expect("register single error");

        println!("cache signal");

        let _ = client.flush();
        break;
    }
}
